package com.iot.app.flink.Source;


import org.apache.log4j.Logger;
import com.iot.app.flink.vo.RedisProp;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.exceptions.JedisConnectionException;

import java.util.Map;

public class RichSource {
    private static RedisProp redisProp;

    public static void initSource(RedisProp prop) {

        if (prop == null) {
            throw new NullPointerException("redis prop is null");
        }

        redisProp = prop;
    }

    public static SourceFunction<Map<String,String>> RedisSouce() {
        return new SourceFunction<Map<String, String>>() {

            private Jedis jedis;
            private boolean isRunning = true;
           // private Logger logger = Logger.getRootLogger();

            @Override
            public void run(SourceContext<Map<String, String>> ctx) throws Exception {
                jedis = new Jedis(redisProp.getHost(), redisProp.getPort());

                while (isRunning) {
                    try {
                        Map<String, String> poi = jedis.hgetAll("POI");
                        if (poi.size() > 0) {
                            ctx.collect(poi);
                        }
                        //Thread.sleep(redisProp.getInterval());
                        Thread.sleep(1000);
                    } catch (JedisConnectionException e) {
                        //logger.error("redis连接异常" + e);
                        this.jedis = new Jedis(redisProp.getHost(), redisProp.getPort());
                    } catch (Exception e) {
                        //logger.error("数据源异常" + e);
                    }
                }
            }

            @Override
            public void cancel() {
                isRunning = false;

                if (jedis != null) {
                    jedis.close();
                }
            }
        };
    }
}
